package com.sunsharing.transport.netty;

import java.net.InetSocketAddress;
import java.util.Iterator;
import java.util.concurrent.atomic.AtomicLong;

import com.sunsharing.common.Constant;
import com.sunsharing.common.Context;
import com.sunsharing.common.ContextCreate;
import com.sunsharing.common.URL;
import com.sunsharing.common.config.Config;
import com.sunsharing.transport.Channel;
import com.sunsharing.transport.ConnectClients;
import com.sunsharing.transport.IoHandle;
import com.sunsharing.transport.LongChannel;
import com.sunsharing.transport.mina.MinaChannel;
import com.sunsharing.transport.mina.SearchBySession;
import org.apache.log4j.Logger;
import org.jboss.netty.channel.ChannelEvent;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelState;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
/**
 * Created with IntelliJ IDEA.
 * User: criss
 * Date: 13-4-23
 * Time: 下午2:39
 * To change this template use File | Settings | File Templates.
 */
public class ObjectEchoServerHandler extends SimpleChannelUpstreamHandler {

    private static final Logger logger = Logger.getLogger(
            ObjectEchoServerHandler.class.getName());

    private final AtomicLong transferredMessages = new AtomicLong();

    public IoHandle ioHandle;
    public URL url;

    public ObjectEchoServerHandler(IoHandle ioHandle,URL url)
    {
        this.ioHandle = ioHandle;
        this.url = url;
    }

    public long getTransferredMessages() {
        return transferredMessages.get();
    }

    @Override
    public void handleUpstream(
            ChannelHandlerContext ctx, ChannelEvent e) throws Exception {
//        if (e instanceof ChannelStateEvent &&
//                ((ChannelStateEvent) e).getState() != ChannelState.INTEREST_OPS) {
//            logger.info(e.toString());
//        }
        super.handleUpstream(ctx, e);
    }

    @Override
    public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception
    {
        logger.error("NETTY已连接!"+ e.getChannel().getRemoteAddress());
        //返回登录信息
        e.getChannel().write(ContextCreate.createConnectContext(url.toString()));
    }

    public void channelClosed(org.jboss.netty.channel.ChannelHandlerContext ctx,
                              org.jboss.netty.channel.ChannelStateEvent e)
            throws java.lang.Exception
    {
        NettyChannel channel = null;
        try
        {
            InetSocketAddress add = ((InetSocketAddress)e.getChannel().getRemoteAddress());
            channel = (NettyChannel)searchChannelBySession(e.getChannel());
            //ioHandle.doclose(channel);
        }catch (Exception e2)
        {
            logger.error("关闭出现异常",e2);
        } finally {
            ConnectClients.removeClient(channel);
        }
        logger.info("{}的连接已关闭!"+ e.getChannel().getRemoteAddress());
    }



    @Override
    public void messageReceived(
            ChannelHandlerContext ctx, MessageEvent e) {

        transferredMessages.incrementAndGet();
        InetSocketAddress add = (InetSocketAddress)e.getRemoteAddress();
        Context con = (Context)e.getMessage();
        if(con.getMsgType()== Constant.MSG_TYPE_CONNECT)
        {
            //收到客户端的汇报身份
            URL returnUrl =
                    LongChannel.initConnectURL(con.getMsg().msg, this.url, false, add.getHostName(), add.getPort());
            NettyChannel channel = null;
            channel = new NettyChannel(e.getChannel(),returnUrl);
            channel.setIoHandle(ioHandle);
            Channel ch = ConnectClients.addClient(channel);
            ioHandle.doConnected(ch);
        }else if(con.getMsgType()==Constant.MSG_TYPE_HEART)
        {
            NettyChannel ch = (NettyChannel)searchChannelBySession(e.getChannel()) ;
            if(ch!=null)
            {
                ch.refreshLastActiveTime();
                //发回
                //logger.error("发回心跳报文:"+ch.url);
                ch.send(ContextCreate.createHeartContext());
            }
        }else
        {
            NettyChannel channel = (NettyChannel)searchChannelBySession(e.getChannel());
            channel.receive(con);
            channel.refreshLastActiveTime();
        }

    }

    @Override
    public void exceptionCaught(
            ChannelHandlerContext ctx, ExceptionEvent e) {

        logger.error("Unexpected exception from downstream.",
                e.getCause());

    }

    private static Channel searchChannelBySession(org.jboss.netty.channel.Channel nch)
    {
        for(Iterator<Channel> iter = ConnectClients.activeClients.iterator();iter.hasNext();)
        {
            Channel c = (Channel)iter.next();
            if(c instanceof NettyChannel)
            {
                if(nch==((NettyChannel)c).session)
                {
                    return c;
                }
            }
        }
        return null;
    }
}
